package com.example.demo.rocketmq;

import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.TimeUnit;

public class producer2 {

    /**
     * 事务消息
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        System.out.println("生产者启动开始");
        //创建生成者组
        TransactionMQProducer producer = new TransactionMQProducer("producer-group");
        //设置namesrv 注册中心地址
        producer.setNamesrvAddr("192.168.1.102:9876");
        // 设置发送超时时限为5s，默认30s
        producer.setSendMsgTimeout(30000);

        producer.setTransactionListener(new TransactionListener() {
            //本地事务方法在这里执行
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("executeLocalTransaction");
                System.out.println("executeLocalTransaction 消息："+new String(message.getBody()));
                //未知的话，broker 会回调checkLocalTransaction
                return LocalTransactionState.UNKNOW;
                //本地执行成功，提交消息确认
                //return LocalTransactionState.COMMIT_MESSAGE;

            }

            //该方法是brocker回调，如果事务消息的状态一直没有提交成COMMIT_MESSAGE，brocker会定期回调来检测本地事务消息的状态
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("checkLocalTransaction");
                System.out.println("checkLocalTransaction 消息："+new String(messageExt.getBody()));

                //未知的话，broker 会回调checkLocalTransaction
                //return LocalTransactionState.UNKNOW;
                //回滚消息
                //return LocalTransactionState.ROLLBACK_MESSAGE;
                //本地执行成功，提交消息确认
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        producer.start();
        // 多等待几秒，防止生产者启动未完成就发送消息
        TimeUnit.SECONDS.sleep(10);

        //消息体
        // topic 消息发送的地址
        // body   消息内容
        Message message = new Message("topic1","tag","key", "事务消息数据1".getBytes());
        producer.sendMessageInTransaction(message,null);

    }


}
